package com.bw.medical.realtime.app.dwd;

import com.bw.medical.realtime.common.MedicalCommon;
import com.bw.medical.realtime.util.CreateEnvUtil;
import com.bw.medical.realtime.util.KafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/*
9.4 交易域处方开单支付成功事务事实表
 */

public class DwdTradePrescriptionPaySuc {
    public static void main(String[] args) throws Exception {
        // TODO 1. 表处理环境准备
        StreamExecutionEnvironment env = CreateEnvUtil.getStreamEnv(8085, "medical_dwd_trade_prescription_pay_suc");
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        env.setParallelism(1);

        // TODO 2. 读取数据
        // 2.1 为处方明细表注入水位线
        String topic = MedicalCommon.KAFKA_ODS_TOPIC;
        String groupIdDetail = "medical_dwd_trade_prescription_pay_suc_detail";
        tableEnv.executeSql("create table " + topic + "_detail(" +
                "`database` string,\n" +
                "`table` string,\n" +
                "`type` string,\n" +
                "`data` map<string, string>,\n" +
                "`old` map<string, string>,\n" +
                "`row_time` as TO_TIMESTAMP(`data`['create_time']),\n" +
                "watermark for `row_time` as `row_time` - interval '5' second\n" +
                ")" + KafkaUtil.getKafkaDDL(topic, groupIdDetail));

        // 2.2 为处方支付成功表注入水位线
        String groupIdSuc = "dwd_trade_prescription_pay_suc_suc";
        tableEnv.executeSql("create table " + topic + "_suc(" +
                "`database` string,\n" +
                "`table` string,\n" +
                "`type` string,\n" +
                "`data` map<string, string>,\n" +
                "`old` map<string, string>,\n" +
                "`row_time` as TO_TIMESTAMP(`data`['update_time']),\n" +
                "watermark for `row_time` as `row_time` - interval '5' second\n" +
                ")" + KafkaUtil.getKafkaDDL(topic, groupIdSuc));

        // TODO 3. 筛选处方明细记录
        Table prescriptionDetail = tableEnv.sqlQuery("select \n" +
                "`data`['id'] `id`,\n" +
                "`data`['count'] `count`,\n" +
                "`data`['medicine_id'] `medicine_id`,\n" +
                "`data`['prescription_id'] `prescription_id`,\n" +
                "`row_time`\n" +
                "from `" + topic + "_detail`\n" +
                "where `table` = 'prescription_detail'\n" +
                "and `type` = 'insert'");
        tableEnv.createTemporaryView("prescription_detail", prescriptionDetail);

        // TODO 4. 筛选处方支付成功记录
        Table prescriptionPaySuc = tableEnv.sqlQuery("select \n" +
                "`data`['id'] `id`,\n" +
                "`data`['total_amount'] `total_amount`,\n" +
                "`data`['update_time'] `prescription_pay_suc_time`,\n" +
                "`data`['consultation_id'] `consultation_id`,\n" +
                "`data`['doctor_id'] `doctor_id`,\n" +
                "`data`['patient_id'] `patient_id`,\n" +
                "`row_time`\n" +
                "from `" + topic + "_suc`\n" +
                "where `table` = 'prescription'\n" +
                "and `type` = 'update'\n" +
                "and `data`['status'] = '203'");
        tableEnv.createTemporaryView("prescription_pay_suc", prescriptionPaySuc);

        // TODO 5. 关联获得结果表
        Table resultTable = tableEnv.sqlQuery("select \n" +
                "`pd`.`id`,\n" +
                "`prescription_pay_suc_time`,\n" +
                "`count`,\n" +
                "`medicine_id`,\n" +
                "`prescription_id`,\n" +
                "`total_amount`,\n" +
                "`consultation_id`,\n" +
                "`doctor_id`,\n" +
                "`patient_id`\n" +
                "from `prescription_detail` `pd`, `prescription_pay_suc` `pps` \n" +
                "where `pd`.`prescription_id` = `pps`.`id`  \n" +
                "and `pd`.row_time >= `pps`.row_time - INTERVAL '15' MINUTE \n" +
                "and `pd`.row_time <= `pps`.row_time + INTERVAL '5' SECOND\n");
        tableEnv.createTemporaryView("result_table", resultTable);

        // TODO 6. 写入 Kafka 目标主题
        String sinkTopic = "medical_dwd_trade_prescription_pay_suc";
        tableEnv.executeSql("create table " + sinkTopic + "(\n" +
                "`id` string,\n" +
                "`prescription_pay_suc_time` string,\n" +
                "`count` string,\n" +
                "`medicine_id` string,\n" +
                "`prescription_id` string,\n" +
                "`total_amount` string,\n" +
                "`consultation_id` string,\n" +
                "`doctor_id` string,\n" +
                "`patient_id` string\n" +
                ") " + KafkaUtil.getKafkaSinkDDL(sinkTopic));
        tableEnv.executeSql("insert into " + sinkTopic + " select * from result_table");

    }
}
